#ifndef B3254917_21C3_4F62_BC55_E5FEBE2327E1
#define B3254917_21C3_4F62_BC55_E5FEBE2327E1
/*实现异步工作器*/

#include "buffer.hpp"
#include <mutex>
#include <condition_variable>
#include <thread>
#include <functional>
#include <memory>
#include <atomic>

namespace mian_yang
{
    using Functor = std::function<void(Buffer &)>;

    enum AsyncType
    {
        ASYNC_SAFE,  // 安全状态， 表示缓冲区满了则阻塞， 避免资源耗尽的风险
        ASYNC_UNSAFE // 不考虑资源耗尽的问题， 无限扩容， 常用于测试
    };

    class AsyncLooper
    {
    public:
        using ptr = std::shared_ptr<AsyncLooper>;

        AsyncLooper(const Functor &cb, AsyncType loop_type = AsyncType::ASYNC_SAFE)
            : _stop(false),
              _thread(std::thread(&AsyncLooper::threadEntry, this)),
              _callBack(cb),
              _looper_type(loop_type) /*默认是安全状态， 缓冲区不够了， 就阻塞*/
        {
        }

        ~AsyncLooper()
        {
            stop();
        }

        void stop()
        {
            _stop = true;           // 将推出标志设置为true
            _cond_con.notify_all(); // 唤醒所有的工作线程
            _thread.join();         // 等待工作线程的退出
        }

        void push(const char *data, size_t len)
        {
            // 1、无限扩容 -- 非安全
            // 2、固定大小 -- 生产缓冲区中满了就阻塞
            std::unique_lock<std::mutex> lock(_mutex); // 加锁

            /*条件变量空值， 若缓冲区剩余空间大小大于数据长度， 则可以添加数据*/
            if (_looper_type == AsyncType::ASYNC_SAFE)
            {
                _cond_pro.wait(lock, [&]()
                               { return _pro_buf.WriteAbleSize() >= len; });
            }

            /*能够走下来， 满足了条件， 可以向缓冲区添加数据*/
            _pro_buf.push(data, len);

            /*唤醒消费者对缓冲区中的数据进行处理*/
            _cond_con.notify_one();
        }

    private:
        /*线程的入口函数。对消费者缓冲区中的数据进行处理， 处理完毕后， 初始化缓冲区， 交换缓冲区*/
        void threadEntry()
        {
            while (!_stop)
            {
                /*判断生产缓冲区有没有数据， 有则交换， 无则阻塞*/
                {
                    std::unique_lock<std::mutex> lock(_mutex); // 加锁

                    /*若当前是推出前被唤醒， 或者有数据被唤醒， 则返回真， 继续向下运行， 否则重新陷入休眠*/
                    _cond_con.wait(lock, [&]()
                                   { return _stop || !_pro_buf.empty(); });

                    /*退出标志被设置， 且生产缓冲区已无数据， 这时候再退出， 否则有可能造成生产缓冲区种有数据
                        但是还没有被完全处理。
                    */
                    if (_stop && _pro_buf.empty()) break;
                    _con_buf.swap(_pro_buf); /*交换*/
                    if (_looper_type == AsyncType::ASYNC_SAFE)
                    {
                        _cond_pro.notify_all(); /*唤醒生产者*/
                    }
                } /*花括号是为了设置生命周期， 保证互斥锁自动解锁*/

                /*被唤醒后， 对缓冲区进行数据处理*/
                _callBack(_con_buf); // 处理

                /*初始化消费缓冲区*/
                _con_buf.reset();
            }
        }

    private:
        Functor _callBack; // 具体对缓冲区数据进行处理的回调函数， 由异步工作器使用者提供

    private:
        AsyncType _looper_type;            // 异步器类别
        std::atomic<bool> _stop;           // 是否停止， 停止就是true， 否则就是false
        Buffer _pro_buf;                   // 生产缓冲区
        Buffer _con_buf;                   // 消费缓冲区
        std::mutex _mutex;                 // 互斥锁， 保证线程安全
        std::condition_variable _cond_pro; // 生产者等待队列， 生产缓冲区没有数据， 处理完消费缓冲区的数据后就休眠
        std::condition_variable _cond_con; // 消费者等待队列
        std::thread _thread;               // 异步工作器的工作线程
    };
}

#endif /* B3254917_21C3_4F62_BC55_E5FEBE2327E1 */
